package cn.tedu.micro.demo.canal.config;

import com.alibaba.otter.canal.protocol.CanalEntry;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import top.javatool.canal.client.client.SimpleCanalClient;
import top.javatool.canal.client.factory.EntryColumnModelFactory;
import top.javatool.canal.client.handler.EntryHandler;
import top.javatool.canal.client.handler.MessageHandler;
import top.javatool.canal.client.handler.RowDataHandler;
import top.javatool.canal.client.handler.impl.AsyncMessageHandlerImpl;
import top.javatool.canal.client.handler.impl.RowDataHandlerImpl;
import top.javatool.canal.client.handler.impl.SyncMessageHandlerImpl;
import top.javatool.canal.client.spring.boot.autoconfigure.ThreadPoolAutoConfiguration;
import top.javatool.canal.client.spring.boot.properties.CanalSimpleProperties;

import java.util.List;
import java.util.concurrent.ExecutorService;

@Configuration
@EnableConfigurationProperties(CanalSimpleProperties.class)
@Import(ThreadPoolAutoConfiguration.class)
public class CanalConfiguration {
    @Bean
    public RowDataHandler<CanalEntry.RowData> rowDataHandler() {
        return new RowDataHandlerImpl(new EntryColumnModelFactory());
    }

    // 异步处理器
    @Bean
    @ConditionalOnProperty(value = "canal.async", havingValue = "true", matchIfMissing = true)
    public MessageHandler asyncMessageHandler(
            RowDataHandler<CanalEntry.RowData> rowDataHandler,
            List<EntryHandler> entryHandlers,
            ExecutorService executorService) {
        return new AsyncMessageHandlerImpl(entryHandlers, rowDataHandler, executorService);
    }

    // 同步处理器（方法名不同）
    @Bean
    @ConditionalOnProperty(value = "canal.async", havingValue = "false")
    public MessageHandler syncMessageHandler(
            RowDataHandler<CanalEntry.RowData> rowDataHandler,
            List<EntryHandler> entryHandlers) {
        return new SyncMessageHandlerImpl(entryHandlers, rowDataHandler);
    }

    @Bean(initMethod = "start", destroyMethod = "stop")
    public SimpleCanalClient simpleCanalClient(
            CanalSimpleProperties properties,
            @Qualifier("asyncMessageHandler") MessageHandler messageHandler) {
        String[] server = properties.getServer().split(":");
        return SimpleCanalClient.builder()
                .hostname(server[0])
                .port(Integer.parseInt(server[1]))
                .destination(properties.getDestination())
                .userName(properties.getUserName())
                .password(properties.getPassword())
                .messageHandler(messageHandler)
                .batchSize(properties.getBatchSize())
                .filter(properties.getFilter())
                .timeout(properties.getTimeout())
                .unit(properties.getUnit())
                .build();
    }
}
